/**
 * Copyright © 2018 嘉源锐信. All rights reserved.
 *
 * @Title: MessageConsumerHandler.java
 * @Project: mq-consumer-client
 * @Package: com.huntkey.rx.sceo.mq.consumer.client.config
 * @Description: TODO
 * @author: liuyong0
 * @date: 2018年9月13日 上午10:15:31
 * @version: V1.0
 */
package com.amos.mqclient.consumer.listener;

import com.alibaba.fastjson.JSONObject;
import com.amos.mqclient.constants.MQConstants;
import com.amos.mqclient.consumer.service.MqLogService;
import com.amos.mqclient.pojo.MqMessage;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @ClassName: MessageConsumerHandler
 * @Description: MQ消息处理流程封装，消息不需签收
 * @author: liuyong0
 * @date: 2018年9月13日 上午10:15:31
 */
@Component
public abstract class MessageHandler implements ChannelAwareMessageListener {

    private static final Logger logger = LoggerFactory.getLogger(MessageHandler.class);
    /**
     * 缓存相应队列消息是否需要签收
     */
    private static ConcurrentHashMap<String, AcknowledgeMode> ackMap = new ConcurrentHashMap<>();

    @Autowired
    private MqLogService mqLogService;

    /**
     * @Title:MessageHandler
     * @Description:TODO
     */
    public MessageHandler() {
        super();
    }

    /**
     * @param message void
     * @Title: onMessageRecieved
     * @date: 2018年9月13日 上午10:45:48
     * @Description: 收到消息后进行预处理
     */
    private void onMessageRecieved(MqMessage message, Channel channel) {
        logger.info("----- onMessageRecieved start -----");
        logger.info("收到消息:" + message + ",准备存入日志.");
        this.mqLogService.updateMessageStatus(message, MQConstants.MESSAGE.CONSUMED_STATUS.RECEIVED);
        logger.info("收到消息:" + message + ",已存入日志，更新状态为：" + MQConstants.MESSAGE.CONSUMED_STATUS.RECEIVED);
        logger.info("----- onMessageRecieved end -----");
    }

    /**
     * @param message void
     * @Title: onMessage
     * @date: 2018年9月13日 上午10:46:13
     * @Description: 继承方消息处理
     */
    @Override
    public final void onMessage(Message message, Channel channel) {
        logger.info("----- onMessage start -----");
        String msg = new String(message.getBody());
        msg = msg.replace("\\", "");
        if (!"{".equals(msg.substring(0, 1))) {
            msg = msg.substring(1, msg.length() - 1);
        }
        MqMessage mqMessage = JSONObject.parseObject(msg, MqMessage.class);
        logger.info("----- onMessage MsgTransforToObject : {} -----", mqMessage.toString());

        //前置处理
        this.onMessageRecieved(mqMessage, channel);
        //消息处理
        //消息处理结果标记
        boolean flag = true;
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String messageId = mqMessage.getMessageId();
        if (!StringUtils.isEmpty(messageId)) {
            MqMessage oldMessage = this.mqLogService.getMessageById(messageId);
            //旧消息，检查上次处理状态
            if (null != oldMessage) {
                String status = oldMessage.getMessageStatus();
                if (MQConstants.MESSAGE.CONSUMED_STATUS.CONSUME_FAILED.equals(status) || MQConstants.MESSAGE.CONSUMED_STATUS.RECEIVED.equals(status)) {
                    //旧消息未消费或者消费失败，继续消费
                    try {
                        this.messageHandler(msg);
                    } catch (Exception e) {
                        flag = false;
                        logger.error("继续消费,调用方处理消息出现异常：" + e.getMessage());
                        this.mqLogService.updateMessageError(messageId, e.getMessage());
                        throw new RuntimeException(e.getMessage());
                    }
                } else if (MQConstants.MESSAGE.CONSUMED_STATUS.SUBMIT_FAILED.equals(status)) {
                    //旧消息未签收，执行签收操作
                    flag = true;
                } else if (MQConstants.MESSAGE.CONSUMED_STATUS.SUBMITTED.equals(status)) {
                    //旧消息已签收，为重复消息，查看ID生成服务是否出错
                    logger.error("旧消息{}已签收，为重复消息，查看ID生成服务是否出错.", messageId);
                } else {
                    logger.error("未知消息{}状态{}.", messageId, status);
                }
            } else {
                //新消息
                try {
                    this.messageHandler(msg);
                } catch (Exception e) {
                    flag = false;
                    logger.error("新消息,调用方处理消息出现异常：" + e.getMessage());
                    this.mqLogService.updateMessageError(messageId, e.getMessage());
                    throw new RuntimeException(e.getMessage());
                }
            }
        } else {
            logger.error("消息{}不和规范，消息id为空", msg);
        }

        //后置处理
        String queue = message.getMessageProperties().getConsumerQueue();
        this.onMessageCompeleted(mqMessage, queue, channel, deliveryTag, flag);
        logger.info("----- onMessage end -----");
    }

    /**
     * @param message
     * @Title: encasedOnMessage
     * @date: 2018年9月13日 上午10:46:00
     * @Description: 处理步骤结合、统一
     */
    public abstract void messageHandler(String message) throws Exception;

    /**
     * @param message
     * @Title: onMessageCompeleted
     * @date: 2018年9月13日 上午10:45:56
     * @Description: 消息处理完毕后进行后置处理
     */
    private void onMessageCompeleted(MqMessage message, String queue, Channel channel, long deliveryTag, boolean flag) {
        logger.info("----- onMessageCompeleted start -----");
        logger.info("消息: " + message + ",处理完成，待进行事务提交、状态更新.");

        String status = MQConstants.MESSAGE.CONSUMED_STATUS.CONSUMED;
        if (flag) {
            AcknowledgeMode ack = ackMap.get(queue);
            if (ack.isManual()) {
                //重试5次
                int retryTimes = 5;
                //进行消息
                RetryTemplate oRetryTemplate = new RetryTemplate();
                SimpleRetryPolicy oRetryPolicy = new SimpleRetryPolicy();
                oRetryPolicy.setMaxAttempts(retryTimes);
                oRetryTemplate.setRetryPolicy(oRetryPolicy);
                try {
                    // obj为doWithRetry的返回结果，可以为任意类型
                    Integer result = oRetryTemplate.execute(new RetryCallback<Integer, Exception>() {
                        int count = 0;

                        @Override
                        public Integer doWithRetry(RetryContext context) throws Exception {//开始重试
                            channel.basicAck(deliveryTag, false);
                            logger.info("消息{}已签收", message.getMessageId());
                            return ++count;
                        }
                    }, new RecoveryCallback<Integer>() {
                        @Override
                        public Integer recover(RetryContext context) throws Exception { //重试多次后都失败了
                            logger.info("消息{}签收失败", message.getMessageId());
                            return Integer.MAX_VALUE;
                        }
                    });

                    if (result.intValue() <= retryTimes) {
                        //消息签收成功
                        status = MQConstants.MESSAGE.CONSUMED_STATUS.SUBMITTED;
                    } else {
                        //MQ服务器或网络出现问题，签收失败
                        status = MQConstants.MESSAGE.CONSUMED_STATUS.SUBMIT_FAILED;
                    }
                } catch (Exception e) {
                    status = MQConstants.MESSAGE.CONSUMED_STATUS.SUBMIT_FAILED;
                    logger.error("消息{}签收出现异常：" + e.getMessage(), message.getMessageId());
                }
            } else {
                logger.info("消息{}自动签收", message.getMessageId());
            }
        } else {
            //消息未被正常消费,不需签收，待再次接收消息进行处理
            status = MQConstants.MESSAGE.CONSUMED_STATUS.CONSUME_FAILED;

        }
        //更新消息状态
        this.mqLogService.updateMessageStatus(message, status);
        logger.info("----- onMessageCompeleted end -----");
    }

    /**
     * @param ack
     * @Title: setAck
     * @date: 2018年9月14日 上午11:17:41
     * @Description: 注入消息签收模式
     */
    public final void setAck(String queue, AcknowledgeMode ack) {
        ackMap.put(queue, ack);
        logger.info("注入队列{}消息签收模式{}", queue, ack.name());
    }
}
